package com.example.test_1_21.mqserver.core;

import com.example.test_1_21.common.Consumer;
import com.example.test_1_21.common.ConsumerEnv;
import com.example.test_1_21.common.MQException;
import com.example.test_1_21.mqserver.VirtualHost;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

//通过这个类来实现消费消息的核心逻辑
public class ConsumerManager {
    //持有上层的VirtualHost对象的引用，用来操作数据
    private VirtualHost parent;
    //指定一个线程池，负责去执行具体的回调任务
    private ExecutorService workerPool= Executors.newFixedThreadPool(4);
    //存放令牌的队列
    private BlockingQueue<String> tokenQueue=new LinkedBlockingDeque<>();
    //扫描线程
    private Thread scannerThread=null;

    public ConsumerManager(VirtualHost p){
        parent=p;
        scannerThread=new Thread(()->{
            while (true){
                try {
                    //1、拿到令牌
                    String queueName=tokenQueue.take();
                    //2、根据令牌找到队列
                    MSGQueue queue=parent.getMemoryDataCenter().getQueue(queueName);
                    if(queue==null){
                        throw new MQException("[ConsumerManager] 取令牌后发现，该队列名不存在！queueName="+queueName);
                    }
                    //3、从这个队列中消费一个消息
                    synchronized (queue){
                        consumeMessage(queue);
                    }
                }catch (InterruptedException | MQException e){
                    e.printStackTrace();
                }
            }
        });
        //把现场设为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MQException {
        MSGQueue queue=parent.getMemoryDataCenter().getQueue(queueName);
        if(queue==null){
            throw new MQException("[ConsumerManager] 队列不存在！queueName="+queueName);
        }
        ConsumerEnv consumerEnv=new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
        synchronized (queue){
            queue.addConsumerEnv(consumerEnv);
            //如果当前队列中已经有了一些消息，需要立即消费掉
            int n=parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                consumeMessage(queue);
            }
        }
    }

    private void consumeMessage(MSGQueue queue) {
        //1、轮询的方式，找个消费者
        ConsumerEnv luckyDog=queue.chooseConsumer();
        if(luckyDog==null){
            return;
        }
        //2、从队列中取出一个消息
        Message message=parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message==null){
            return;
        }
        //3、把消息带入到消费者的回调方法中，丢给线程池执行
        workerPool.submit(()->{
            try {
                //1、把消息放到待确认的集合中，这个操作必在执行回调前
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(),message);
                //2、真正执行回调操作
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(),message.getBasicProperties(),message.getBody());
                //3、如果当前“自动应答”，就可以直接把消息删除；如果是“手动应答”，先不处理，交给后续消费者调用basicAck处理
                if(luckyDog.isAutoAck()){
                    //删除硬盘上的消息
                    if(message.getDeliverMode()==2){
                        parent.getDiskDataCenter().deleteMessage(queue,message);
                    }
                    //删除上面的待确认集合中的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(),message.getMessageId());
                    //删除内存中消息中心的消息
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息被成功消费！queueName="+queue.getName());
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        });
    }
}
